package com.per.mybootall.es.canal;


import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import com.per.mybootall.common.TableEnum;
import com.per.mybootall.config.SpringContextHolder;
import com.per.mybootall.es.ElasticSearchHighUtils;
import com.per.mybootall.es.ElasticSearchHighUtilsTestThread;
import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created with IntelliJ IDEA.
 *
 * @author: yyy
 * Date: 2022/5/10
 * Time: 17:36
 * Description:
 */
@Component
public class CanalListerner {
    @Resource
    private RestHighLevelClient restHighLevelClient;

    Logger log= LoggerFactory.getLogger(CanalListerner.class);
    public void afterPropertiesSet() throws InterruptedException, IOException {
        System.out.println("开始监听==========》");
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("150.158.98.73", 11111),
                "example", "", "");
        while (true){
            connector.connect();
            connector.subscribe("yyy_test_canal.*");
            Message message =connector.get(100);
            List<CanalEntry.Entry> messageEntry=message.getEntries();
            if(messageEntry.size()<=0){
                Thread.sleep(1000);
            }else{
                for(CanalEntry.Entry entry:messageEntry){
                  String tableName=entry.getHeader().getTableName();
                    CanalEntry.EntryType  entryType= entry.getEntryType();
                    ByteString StoreValue= entry.getStoreValue();
                    if(CanalEntry.EntryType.ROWDATA.equals(entryType)){
                       CanalEntry.RowChange rowChange= CanalEntry.RowChange.parseFrom(StoreValue);
                        CanalEntry.EventType eventType= rowChange.getEventType();
                        List<CanalEntry.RowData> rowDataList=rowChange.getRowDatasList();
                        for(CanalEntry.RowData rowData:rowDataList){

                            JSONObject beforeData=new JSONObject();
                            List<CanalEntry.Column> beforeColumnsList=rowData.getBeforeColumnsList();
                            for(CanalEntry.Column column:beforeColumnsList){
                                beforeData.put(column.getName(),column.getValue());
                            }

                            JSONObject afterData=new JSONObject();
                            List<CanalEntry.Column> afterColumnsList=rowData.getAfterColumnsList();
                            for(CanalEntry.Column column:afterColumnsList){
                                afterData.put(column.getName(),column.getValue());
                            }
                            System.out.println("数据打印：表名："+tableName+"             事件类型:"+eventType
                                    +"       之前数据："+beforeData+"        之后数据："+afterData);
                            if(TableEnum.getEnumByName(tableName)!=null){
                                matchingType(tableName,eventType,beforeData,afterData);
                                System.out.println("over----------------------------------------------");
                            }
                        }
                    }else{
                        System.out.println("当前类型为："+entryType);
                    }
                }
            }
        }
    }
    //这一步尽量换到消息队列，不然玩意执行失败就没发继续执行了
    @Async
    public void matchingType(String tableName,CanalEntry.EventType eventType,JSONObject beforeData,JSONObject afterData) throws IOException {
        System.out.println("--------------------------->>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
        SpringContextHolder springContextHolder = new SpringContextHolder();
        RestHighLevelClient restHighLevelClient = springContextHolder.getBean("restHighLevelClient", RestHighLevelClient.class);

        ElasticSearchHighUtils elasticSearch=new ElasticSearchHighUtils();

        if(!elasticSearch.testIndexExist(restHighLevelClient,tableName)){
            //elasticSearch.creatIndex(restHighLevelClient,tableName);
            elasticSearch.creatMapping(restHighLevelClient,tableName);
        }
        Map afterMap=JSONObject.parseObject(afterData.toString(), HashMap.class);
        switch (eventType){
            case INSERT:
                elasticSearch.testAddDocument(restHighLevelClient,afterMap,tableName);
                break;
            case UPDATE:
                //elasticSearch.testUpdateRequest(restHighLevelClient,afterMap,tableName);
                break;
            case DELETE:
                Map beforeMap=JSONObject.parseObject(beforeData.toString(), HashMap.class);
                elasticSearch.testDelectDocument(restHighLevelClient,beforeMap,tableName);
                break;
            default:
                log.error("不支持的sql类型"+"  表名:"+tableName+"  事件类型:"+eventType+"   sql执行之前数据"+beforeData+"    sql执行之后数据:"+afterData);
        }

        //查询es相关结果集
//        if(!eventType.equals("DELETE")){
//            elasticSearch.testSearch(restHighLevelClient,afterMap,tableName);
//        }
    }

}
